package com.yixiu.bi.kafka;

import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.*;


/**
 * Kafka消费者客户端工具
 *
 *建议：所有的配置文件路径，写到一个枚举类中。
 *
 * 注意：
 * (1)
 *
 *
 * 参考：
 *(1)kafka java api    http://www.cnblogs.com/qizhelongdeyang/p/7355309.html
 *
 */
public class KafkaConsumerUtil {
    private  static Logger logger  = LoggerFactory.getLogger(KafkaConsumerUtil.class);

    // 使用这个静态变量保存每一个配置文件对应的kafaka消费者
    private static ConcurrentHashMap<String,KafkaConsumerUtil> kafkaUtilMap  = new ConcurrentHashMap<String, KafkaConsumerUtil>();

    private String topic = ""; // topic
    private KafkaConsumer consumer = null; // 消费者
    private  String configFile = ""; // 配置文件
    private ExecutorService executor; // 通过线程池实现多线程消费消息
    private  long  consumeTimeout ; //消费消息超时设置

    /**
     * 根据配置文件初始化KafkaConsumerUtil对象
     * @param configFile 配置文件路径
     */
    private KafkaConsumerUtil(String configFile) {
        this.configFile = configFile;
        // 读取配置，初始化consumer 、topic、executor

        try {
            Properties kafkaProperties = new Properties();
            kafkaProperties.load(KafkaConsumerUtil.class.getResourceAsStream(configFile));
            // 读取topic
            this.topic  = kafkaProperties.getProperty("topic");
            if(StringUtils.isEmpty(this.topic)) {
                logger.error( configFile + "没有配置topic");
                return;
            }
            // 用于建立与kafka集群的连接，这个list仅仅影响用于初始化的hosts，来发现全部的servers。
            //格式：host1:port1,host2:port2,…，数量尽量不止一个，以防其中一个down了
            String broker = kafkaProperties.getProperty("bootstrap.servers");
            if(StringUtils.isEmpty(broker)) {
                logger.error(configFile + "没有配置bootstrap.servers");
                return;
            }
            // 其他的配置信息，均采用Kafka客户端默认配置
            // 同一组的消费者共享offset,如果有两组消费者,它们分别按照自己的offset消费消息，可能都从头消费
            if(!kafkaProperties.containsKey("group.id")) { // 默认 group1
                kafkaProperties.put("group.id","group1");
            }
            // 反序列化的默认类
            if(!kafkaProperties.containsKey("key.deserializer")) {
                kafkaProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            }
            if(!kafkaProperties.containsKey("value.deserializer")) {
                kafkaProperties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            }


            // 默认线程池为1
            int threadCount  = Integer.valueOf(kafkaProperties.getProperty("thread.count","1"));
            executor = Executors.newFixedThreadPool(threadCount);

            this.consumeTimeout = Long.valueOf(kafkaProperties.getProperty("consumer.poll.timeout","3000"));


            // 初始化KafkaProducer，有可能异常
            consumer  = new KafkaConsumer(kafkaProperties);
            consumer.subscribe(Arrays.asList(this.topic)); // 订阅topic

        }catch (IOException ex) {
            logger.error("配置文件不存在:" + configFile,ex);
        } catch (Exception ex) {
            logger.error("创建KafkaProducerUtil异常:" + configFile,ex);
            this.consumer = null;
        }
    }


    /**
     * 根据配置文件获取工具
     * kafkaConsumerUtil.producer != null 说明获取成功，否则失败
     * @param configFile
     * @return kafkaConsumerUtil.producer != null 说明获取成功，否则失败
     */
    private static KafkaConsumerUtil getKafkaProducerUtil(String configFile) {
        if(kafkaUtilMap.containsKey(configFile)) {
            return kafkaUtilMap.get(configFile);
        }
        KafkaConsumerUtil kafkaConsumerUtil  = new KafkaConsumerUtil(configFile);
        if(kafkaConsumerUtil.consumer != null) {
            synchronized (kafkaUtilMap) {
                if(!kafkaUtilMap.containsKey(configFile)) {
                    kafkaUtilMap.put(configFile,kafkaConsumerUtil);
                }
            }
        }
        return  kafkaConsumerUtil;
    }


    /**
     *  设置kafka 消费者的监听程序
     *  外部调用，只需要new 一个KafkaConsumerListener对象，重写onMessage方法即可
     * @param configFile
     * @param kafkaConsumerListener
     */
    public static void setKafkaConsumerListener(String configFile, KafkaConsumerListener kafkaConsumerListener) {

        KafkaConsumerUtil kafkaConsumerUtil = KafkaConsumerUtil.getKafkaProducerUtil(configFile);
        // 主线程负责消费消息(通过死循环实现,没有类似于 consumer.hasNext 的方法)
        while (true) {
            // 每一次拉取一组消息
            ConsumerRecords<String, String> records = kafkaConsumerUtil.consumer.poll(kafkaConsumerUtil.consumeTimeout);

            for (ConsumerRecord<String, String> record : records) {

               KafkaConsumerListener.ConsumerThreadHandler consumerThreadHandler = kafkaConsumerListener.newConsumerThreadHandler(record);
               // 多线程负责处理消息(将线程放入线程池)
               kafkaConsumerUtil.executor.submit(consumerThreadHandler);
            }
        }

    }


}
